package com.example.rabbitdemo;

import android.os.Bundle;
import android.os.Handler;
import android.os.Message;
import android.util.Log;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.TimeoutException;

public class RabbitMqUtils {
    private static Connection connection;
    private static Channel channel;
    static int i=0;

    public static ConnectionFactory setUpConnectionFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.IpHost);//主机地址：
        factory.setPort(Constants.Port);// 端口号
        factory.setUsername(Constants.username);// 用户名
        factory.setPassword(Constants.password);// 密码

        factory.setAutomaticRecoveryEnabled(true);// 设置连接恢复
        factory.setRequestedHeartbeat(5);
        factory.setNetworkRecoveryInterval(2000);
        return factory;

    }

    static Thread subscribe(final Handler handler) {
        ConnectionFactory factory = setUpConnectionFactory();
        Thread subThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    // 创建新的连接
                    connection = factory.newConnection();
                    // 创建通道
                    channel = connection.createChannel();
                    // 处理完一个消息，再接收下一个消息
                    channel.basicQos(1);
                    // 随机命名一个队列名称
                    String queueName = System.currentTimeMillis() + "queueName";
                    // 声明交换机类型
                    channel.exchangeDeclare(Constants.MQ_EXCHANGE_CAR, "topic", true);
                    Log.i("topic", "run: " + Constants.MQ_EXCHANGE_CAR);
                    // 声明队列（持久的、非独占的、连接断开后队列会自动删除）
                    AMQP.Queue.DeclareOk q = channel.queueDeclare(queueName, false, false, true, null);// 声明共享队列
                    // 根据路由键将队列绑定到交换机上（需要知道交换机名称和路由键名称）
                   channel.queueBind(q.getQueue(), Constants.MQ_EXCHANGE_CAR, Constants.routingKey);
                    //创建消费者
                    Consumer consumer = new DefaultConsumer(channel) {

                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                            super.handleDelivery(consumerTag, envelope, properties, body);
                            String msg = new String(body);
                            Message message = handler.obtainMessage();
                            Bundle bundle = new Bundle();
                            bundle.putString("msg", msg);
                            message.setData(bundle);
                            handler.sendMessage(message);
                            Log.i("topic", "run: " + msg);
                        }

                    };
                    channel.basicConsume(q.getQueue(), true, consumer);


                } catch (IOException e) {
                    e.printStackTrace();
                    Log.e("IOException", "run: " + e.toString());
                } catch (TimeoutException e) {
                    e.printStackTrace();
                    Log.e("TimeoutException", "run: " + e.toString());
                }
            }

        });
          subThread.start();
          return  subThread;
    }

    public static void closeCoon(Thread thread) {

        Thread closeThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    if (connection != null && connection.isOpen())
                        connection.close();
                    if (channel != null && channel.isOpen())
                        channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        });
        closeThread.start();
        closeThread.interrupt();
        if (thread != null) thread.interrupt();
    }

    public static void publishToAMPQ( BlockingDeque<String> queue) {
        ConnectionFactory factory = setUpConnectionFactory();

        Thread publishThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        // 创建连接
                        Connection connection = factory.newConnection();
                        // 创建通道
                        Channel ch = connection.createChannel();
                        ch.confirmSelect();
                        while (true) {
                            String message = queue.takeFirst();
                            try {
                                // 发布消息
                                ch.basicPublish(Constants.MQ_EXCHANGE_CAR, Constants.routingKey, null, message.getBytes());
                                ch.waitForConfirmsOrDie();

                            } catch (Exception e) {
                                queue.putFirst(message);
                                throw e;
                            }

                        }
                    } catch (InterruptedException e) {
                        break;
                    } catch (Exception e) {
                        Log.d("TAG_Publish", "Connection broken: " + e.getClass().getName());
                        try {
                            Thread.sleep(5000); //sleep and then try again
                        } catch (InterruptedException e1) {
                            break;

                        }

                    }

                }

            }

        });
        publishThread.start();
    }
    static void publishMessage(BlockingDeque<String> queue) {
        //向内部阻塞队列添加一条消息
        try {

            queue.putLast("test++++++++++");
            Log.i("Service", "run: "+(i++));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

}
